// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin
#pragma once

#include <algorithm>
#include <map>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/feed_util/book_builder.h"
#include "coin2/exchange/base/feed_util/nav.h"
#include "coin2/exchange/base/feed_util/kline.h"
#include "coin2/exchange/base/feed_util/iv_greeks.h"
#include "coin2/exchange/base/feed_util/wallet.h"
#include "coin2/exchange/base/feed_util/market_analysis.h"
#include "coin2/exchange/base/feed_util/trading_data.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/feed/fastfeed/fastfeed_reader.h"
#include "coin2/feed/universe/fastfeed_universe_reader.h"

namespace coin2::feed::fastfeed {
namespace impl {

using coin2::base::config::ValidateArchiveConfig;
using coin2::exchange::base::market::ExchangeType;
using coin2::exchange::base::market::MarketType;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::feed::ArchiveConfig;
using coin2::exchange::feed::FeedSystemConfig;
using coin2::exchange::base::feed_util::Index;
using ::fastfeed::proto::Feed;

const double kDefaultMultiplier = std::pow(0.1, 9);

// TODO(jshin) FeedUpdate inherits TopicRecordUpdate. it has nothing to do with fastfeed

struct PqMultiplier {
  double price_multiplier;
  double qty_multiplier;
};

struct FastFeedUpdatePortableElem {
  std::shared_ptr<BookBuilder> book_;
  Trade trade_{};
  Feed feed_;
  std::unordered_map<int, std::string> index_to_norm_symbol_;
  LiquidationOrder liquidation_order_{};
  OpenInterest open_interest_{};
  Index index_{};
  MarkPrice mark_price_{};
  FundingRate funding_rate_{};
  IvGreeks iv_greeks_{};
  Nav nav_{};
  Wallet wallet_{};
  MarketAnalysis market_analysis_{};
};

class FastFeedUpdate : public exchange::base::feed::FeedUpdate {
 public:
  FastFeedUpdate(const std::string& system_name, const MarketExchangeApi& mea)
      : FeedUpdate(system_name, mea) {}

  // Not guaranteed to work. please check before use.
  // should be slow, heavy, let's use it for research purpose only.
  // though fastfeed can't be the production one..
  std::unique_ptr<FastFeedUpdate> Copy() const {
    std::unique_ptr<FastFeedUpdate> ret(new FastFeedUpdate(this->system_name(), this->mea()));
    ret->product_ = this->product_;
    ret->feed_type_ = this->feed_type_;
    ret->native_symbol_ = this->native_symbol_;
    ret->pqmults_ = this->pqmults_;
    ret->SetTimestamp(this->timestamp());
    ret->container_.reset(new FastFeedUpdatePortableElem());
    if (this->is_book()) {
      ret->container_->book_.reset(new BookBuilder(*this->GetBookBuilder(), 20));
      ret->container_->book_->SetTimestamp(this->GetBookBuilder()->Timestamp());
      ret->SetBook(ret->container_->book_.get());
    } else if (this->is_trade()) {
      ret->container_->trade_ = this->trade();
      ret->SetTrade(&ret->container_->trade_);
    } else if (this->is_liquidation()) {
      ret->container_->liquidation_order_ = this->liquidation_order();
      ret->SetLiquidationOrder(&ret->container_->liquidation_order_);
    } else if (this->is_open_interest()) {
      ret->container_->open_interest_ = this->open_interest();
      ret->SetOpenInterest(&ret->container_->open_interest_);
    } else if (this->is_wallet()) {
      ret->container_->wallet_ = this->wallet();
      ret->SetWallet(&ret->container_->wallet_);
    } else if (this->is_market_analysis()) {
      ret->container_->market_analysis_ = this->market_analysis();
      ret->SetMarketAnalysis(&ret->container_->market_analysis_);
    } else if (this->is_mark_price()) {
      ret->container_->mark_price_ = this->mark_price();
      ret->SetMarkPrice(&ret->container_->mark_price_);
    } else if (this->is_index()) {
      ret->container_->index_ = this->index();
      ret->SetIndex(&(ret->container_->index_));
    } else if (this->is_funding_rate()) {
      ret->container_->funding_rate_ = this->funding_rate();
      ret->SetFundingRate(&ret->container_->funding_rate_);
    }
    return ret;
  }

  void SetPriceQtyMultipliers(
      const std::string& norm_symbol,
      double price_multiplier,
      double qty_multiplier) {
    pqmults_[norm_symbol] = PqMultiplier{ price_multiplier, qty_multiplier };
  }

  const PqMultiplier* GetPqMultipliers(const std::string& norm_symbol) const {
    if (pqmults_.count(norm_symbol) == 0) {
      return nullptr;
    }
    return &pqmults_.at(norm_symbol);
  }

 protected:
  std::map<std::string, PqMultiplier> pqmults_;
  mutable std::unique_ptr<FastFeedUpdatePortableElem> container_;
};

inline const std::string& value_or(const std::string& lhs, const std::string& rhs) {
  if (!lhs.empty()) {
    return lhs;
  }
  return rhs;
}
inline int64_t value_or(int64_t lhs, int64_t rhs) {
  if (lhs != 0) {
    return lhs;
  }
  return rhs;
}

inline coin2::exchange::feed::TradeSide ConvertTradeSide(::fastfeed::proto::TradeSide side) {
  switch (side)
  {
  case ::fastfeed::proto::TRADE_SIDE_INVALID:
    return coin2::exchange::feed::TRADE_UNKNOWN_SIDE;
    break;
  case ::fastfeed::proto::TRADE_SIDE_BUY:
    return coin2::exchange::feed::TRADE_BUY_SIDE;
    break;
  case ::fastfeed::proto::TRADE_SIDE_SELL:
    return coin2::exchange::feed::TRADE_SELL_SIDE;
    break;
  case ::fastfeed::proto::TRADE_SIDE_UNKNOWN:
    return coin2::exchange::feed::TRADE_NO_SIDE;
    break;

  default:
    NOTREACHED() << "Invalid trade side " << side;
    break;
  }

  return coin2::exchange::feed::TRADE_UNKNOWN_SIDE;
}

// 1 FastFeedSubsystem = 1 mear
// so we can control root_dir, machine, etc per exchange.
class FastFeedSubsystem : public exchange::base::feed::IFeedSubsystem,
                          public exchange::base::feed::FeedArchiveMixin {
 public:
  FastFeedSubsystem(
      const std::string& system_name,
      const MarketExchangeApiRecipe& mear,
      const coin2::exchange::di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      exchange::base::executor::ArchiveSyncExecutor* ass,
      exchange::base::feed::IFeedSubscriber* callback)
      : IFeedSubsystem(system_name, mear.mea), FeedArchiveMixin(ass), mear_(mear), config_(config) {
    AddCallback(callback);
    CHECK_EQ_THROW(
        config.archive().archive_type(),
        coin2::exchange::feed::ArchiveConfig_ArchiveType_FASTFEED);
    ValidateArchiveConfig(config.archive());

    const auto& archive = config_.archive();
    auto interval = coin2::base::config::FromIntervalConfig(archive.interval());
    auto warmup = DurationFromString(archive.read_warmup_duration());

    LOG(INFO) << "[FastFeed System]";
    LOG(INFO) << "              mea: " << mea_.String();
    LOG(INFO) << "           recipe: " << mear_.recipe;
    LOG(INFO) << "interval - warmup: " << DebugStringFromDuration(warmup);
    LOG(INFO) << "interval -  start: " << Iso8601FromTimestamp(interval.first);
    LOG(INFO) << "interval -    end: " << Iso8601FromTimestamp(interval.second);

    // FastFeedReader is capable of handling multiple exchanges,
    // but here we use it for one exchange
    start_time_ = interval.first;
    reader_ = std::make_unique<FastFeedReader>(
        mear_,
        config_.products(),
        archive.machine(),
        archive.fastfeed_root_dir(),
        archive.worker(),
        archive.worker2(),
        warmup,
        interval.first,
        interval.second,
        archive.max_record_size());

    update_ptr_.reset(new FastFeedUpdate(system_name, mea_));
  }

  //     feed_type: Book,
  //     data_type: BookSnapshot,
  //     symbol: "ADA-ETH.Huobi.adaeth",

  void Prepare() override {}
  void Cleanup() override {}

  bool BanProduct(const IProduct* product) final {
    LOG(INFO) << "Unsubscribing " << product->relative_norm();
    return reader_->BanProduct(product);
  }

  bool BanProduct(const std::vector<const IProduct*>& products) final {
    bool ret = false;
    for (auto& product : products) {
      ret |= reader_->BanProduct(product);
    }
    return ret;
  }

  int64_t PeekTimestamp() override {
    if (!has_feed_) {
      while (reader_->Read(&feed_)) {
        feed_.set_timestamp(feed_.timestamp() + config_.archive().offset_ns());
        has_feed_ = true;
        break;
      }
      if (!has_feed_) {
        return 0;
      }
    }
    return feed_.timestamp();
  }

  void ConsumeRecord() override {
    using namespace ::fastfeed::proto;
    CHECK_THROW(has_feed_) << "Cannot consume a feed";
    has_feed_ = false;

    std::string norm_symbol;
    const std::string& fastfeed_symbol = feed_.symbol();
    int dots = std::count(fastfeed_symbol.begin(), fastfeed_symbol.end(), '.');

    // (BTC-USD or BTC-USD.xxx) 0~1 vs BTC-USD.20191206.Okex.BTC1206 2~3
    if (dots <= 1) {
      norm_symbol = fastfeed_symbol;
    } else {
      // coin1
      if (dots == 2) {
        norm_symbol = fastfeed_symbol.substr(0, fastfeed_symbol.find('.'));
      } else {
        if (dots == 3 || dots == 4) {
          int dot2 = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
          int dot3 = fastfeed_symbol.find('.', dot2 + 1);
          auto between = fastfeed_symbol.substr(dot2 + 1, dot3 - dot2 - 1);
          if (between == "P" || between == "C") {
            norm_symbol = fastfeed_symbol;
          } else {
            int dot = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
            norm_symbol = fastfeed_symbol.substr(0, dot);
          }
        } else {
          int dot = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
          norm_symbol = fastfeed_symbol.substr(0, dot);
        }
      }
    }
    FastFeedUpdate& update_ = *update_ptr_.get();
    update_.Reset();
    update_.SetTimestamp(feed_.timestamp());
    update_.SetRecipe(mear_.recipe);
    // clear
    update_.SetKlines(nullptr);

    if (feed_.timestamp() >= start_time_) {
      if (products_.find(norm_symbol) == products_.end()) {
        products_[norm_symbol] = CreateProductFromNormString(
            mea_, norm_symbol, feed_.timestamp());
      }
      update_.SetProduct(*products_[norm_symbol]);
      update_.SetNativeSymbol(products_[norm_symbol]->feed_native_symbol());
    }

    const auto* pqmult = update_.GetPqMultipliers(norm_symbol);

    if (feed_.feed_type() == FEED_TYPE_METADATA) {
      if (feed_.data_type() == DATA_TYPE_BOOK_BUILDER_PARAMETER) {
        const auto& param = feed_.book_builder_param();
        update_.SetPriceQtyMultipliers(
            norm_symbol,
            pow(0.1, param.price_precision()),
            pow(0.1, param.qty_precision()));
      }
    } else if (feed_.feed_type() == FEED_TYPE_BOOK) {
      // not ready yet.
      if (!pqmult) {
        return;
      }
      BookBuilder* b = books_[norm_symbol].get();
      if (b == nullptr) {
        BookBuilder::BookOption book_type{BookBuilder::SNAPSHOT};
        if (config_.raw_parser().enable_diff_book()) {
          book_type = BookBuilder::DELTA;
        } else if (config_.raw_parser().publish_bbo_book_only()) {
          book_type = BookBuilder::LIGHT_BBO;
        }
        books_[norm_symbol].reset(b = new BookBuilder(book_type));
      }

      if (feed_.data_type() == DATA_TYPE_BOOK_SNAPSHOT) {
        const auto& snapshot = feed_.book_snapshot();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        tx.Clear();
        for (const auto& level : snapshot.ask()) {
          tx.UpdateAsk(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
        for (const auto& level : snapshot.bid()) {
          tx.UpdateBid(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
      } else if (feed_.data_type() == DATA_TYPE_BOOK_UPDATE) {
        const auto& diff = feed_.book_update();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        for (const auto& level : diff.ask()) {
          tx.UpdateAsk(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
        for (const auto& level : diff.bid()) {
          tx.UpdateBid(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
      }

      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      update_.SetFeedType(coin2::exchange::feed::BOOK_FEED);
      update_.SetBook(b);
      update_.MutableTs().set_exchange_publish_timestamp(b->Timestamp());
      onBookFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_TRADE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& trade = feed_.trade();
      update_.SetFeedType(coin2::exchange::feed::TRADE_FEED);
      update_.SetTrade(&trade_);
      trade_ = {
          feed_.origin_timestamp(),
          static_cast<double>(trade.price() * pqmult->price_multiplier),
          static_cast<double>(trade.qty() * pqmult->qty_multiplier),
          ConvertTradeSide(trade.side()),
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
      };
      update_.MutableTs().set_exchange_publish_timestamp(trade_.timestamp);
      onTradeFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_LIQUIDATION_ORDER) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& liquidation = feed_.liquidation();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetLiquidationOrder(&liquidation_order_);
      coin::proto::OrderSide order_side
          = coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
      if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_SELL) {
        order_side = coin::proto::OrderSide::SELL_ORDER;
      } else if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_BUY) {
        order_side = coin::proto::OrderSide::BUY_ORDER;
      }
      liquidation_order_ = {
          feed_.origin_timestamp(),
          static_cast<double>(liquidation.price() * pqmult->price_multiplier),
          static_cast<double>(liquidation.qty() * pqmult->qty_multiplier),
          0,
          0,
          coin::proto::OrderType::UNKNOWN_ORDER_TYPE,
          order_side,
          coin::proto::OrderDuration::GTC_ORDER,
          coin::proto::OrderState::CREATED_ORDER,
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
          liquidation.order_id(),
      };
      update_.MutableTs().set_exchange_publish_timestamp(liquidation_order_.timestamp);
      onLiquidationFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_OPEN_INTEREST) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& open_interest = feed_.open_interest();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetOpenInterest(&open_interest_);
      open_interest_ = {
          feed_.origin_timestamp(),
          static_cast<double>(open_interest.qty() * pqmult->qty_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(open_interest_.timestamp);
      onOpenInterestFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_INDEX) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& index = feed_.index();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetIndex(&index_);
      index_ = {
          feed_.origin_timestamp(),
          static_cast<double>(index.price() * pqmult->price_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(index_.timestamp);
      onIndexFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_MARK_PRICE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& mark_price = feed_.mark_price();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetMarkPrice(&mark_price_);
      mark_price_ = {
          feed_.origin_timestamp(),
          static_cast<double>(mark_price.price() * pqmult->price_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(mark_price_.timestamp);
      onMarkPriceFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_FUNDING_RATE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& funding_rate = feed_.funding_rate();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetFundingRate(&funding_rate_);
      funding_rate_ = {
          feed_.origin_timestamp(),
          // coin/fastfeed/feed.proto:  int64 funding_rate = 1;
          static_cast<double>(funding_rate.funding_rate()) * kDefaultMultiplier,
          funding_rate.funding_time(),
          // coin/fastfeed/feed.proto:  int64 estimated_rate = 3;
          static_cast<double>(funding_rate.estimated_rate()) * kDefaultMultiplier,
      };
      update_.MutableTs().set_exchange_publish_timestamp(funding_rate_.timestamp);
      onFundingRateFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_IV_GREEKS) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& iv_greeks = feed_.iv_greeks();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetIvGreeks(&iv_greeks_);
      iv_greeks_ = {
          feed_.origin_timestamp(),
          static_cast<double>(iv_greeks.ask_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.bid_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.mark_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.delta() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.gamma() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.rho() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.theta() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.vega() * kDefaultMultiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(iv_greeks_.timestamp);
      onIvGreeksFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_NAV) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& nav = feed_.nav();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetNav(&nav_);

      std::vector<::UnderlyingAsset> basket;
      for (auto& ua : nav.basket()) {
        basket.emplace_back(std::string(ua.symbol_or_currency()),
                            static_cast<double>(ua.amount() * kDefaultMultiplier));
      }
      nav_ = {
          feed_.origin_timestamp(),
          static_cast<double>(nav.actual_leverage() * kDefaultMultiplier),
          static_cast<double>(nav.nav() * pqmult->price_multiplier),
          static_cast<double>(nav.outstanding() * kDefaultMultiplier),
          basket,
      };
      update_.MutableTs().set_exchange_publish_timestamp(nav_.timestamp);
      onNavFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_KLINE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      Klines* klines = nullptr;
      if (feed_.kline_snapshot().kline_size() > 0) {
        for (const auto& kline : feed_.kline_snapshot().kline()) {
          if (klines == nullptr) {
            klines = &(klines_[std::make_pair(norm_symbol, kline.interval())]);
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetKlines(klines);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            klines->SnapshotClear();
          }

          // fastfeed only save freezed kline
          auto* pKline = klines->AddKline();
          pKline->timestamp = feed_.origin_timestamp();
          pKline->interval = kline.interval();
          pKline->open_timestamp = kline.open_timestamp();
          pKline->close_timestamp = kline.close_timestamp(),
          pKline->open = kline.open();
          pKline->close = kline.close();
          pKline->high = kline.high();
          pKline->low = kline.low();
          pKline->volume = kline.volume();
          pKline->turnover = kline.turnover();
          pKline->buy_volume = kline.buy_volume();
          pKline->buy_turnover = kline.buy_turnover();
        }
      } else if (feed_.kline_update().kline_size() > 0) {
        for (const auto& kline : feed_.kline_update().kline()) {
          if (klines == nullptr) {
            klines = &(klines_[std::make_pair(norm_symbol, kline.interval())]);
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetKlines(klines);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            klines->UpdateClear();
          }

          // fastfeed only save freezed kline
          auto* pKline = klines->AddKline();
          pKline->timestamp = feed_.origin_timestamp();
          pKline->interval = kline.interval();
          pKline->open_timestamp = kline.open_timestamp();
          pKline->close_timestamp = kline.close_timestamp(),
          pKline->open = kline.open();
          pKline->close = kline.close();
          pKline->high = kline.high();
          pKline->low = kline.low();
          if (kline.volume() != 0) pKline->volume = kline.volume();
          if (kline.turnover() != 0) pKline->turnover = kline.turnover();
          if (kline.buy_volume() != 0) pKline->buy_volume = kline.buy_volume();
          if (kline.buy_turnover() != 0) pKline->buy_turnover = kline.buy_turnover();
        }
      } else {
        LOG(INFO) << "No klines data";
        return;
      }

      onKlineFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_WALLET) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& wallet = feed_.wallet();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetWallet(&wallet_);
      wallet_ = {
          feed_.origin_timestamp(),
          wallet.can_deposit(),
          wallet.can_withdraw(),
          wallet.blockchain_delay()
      };
      update_.MutableTs().set_exchange_publish_timestamp(wallet_.timestamp);
      onWalletFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_MARKET) {
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& market_analysis = feed_.market_analysis();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetMarketAnalysis(&market_analysis_);
      std::vector<::VolumePerExchange> volume_per_exchange_vec;
      for (auto& volume_per_exchange : market_analysis.volume_per_exchange_list()) {
        volume_per_exchange_vec.emplace_back(
          volume_per_exchange.exchange_name(),
          volume_per_exchange.volume(),
          volume_per_exchange.volume_in_usd()
        );
      }
      market_analysis_ = {
          feed_.origin_timestamp(),
          market_analysis.total_supply(),
          market_analysis.max_supply(),
          market_analysis.circulating_supply(),
          market_analysis.market_cap_in_usd(),
          market_analysis.tvl_in_usd(),
          market_analysis.total_volume_in_usd(),
          volume_per_exchange_vec
      };
      update_.MutableTs().set_exchange_publish_timestamp(market_analysis_.timestamp);
      onMarketAnalysisFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_TRADING_DATA) {
      if (!pqmult) { return; }
      // skip warmup
      if (update_.timestamp() < start_time_) { return; }

      const ::fastfeed::proto::TradingData* trading_data = nullptr;
      bool is_snapshot = false;
      if (feed_.has_trading_data_snapshot()) {
        is_snapshot = true;
        trading_data = &feed_.trading_data_snapshot();
      } else if (feed_.has_trading_data_update()) {
        trading_data = &feed_.trading_data_update();
      } else {
        THROW() << "No trading data";
      }

      if (trading_data->top_long_short_account_ratio_size() > 0) {
        ::TradingData<TopLongShortAccountRatio>* pratio = nullptr;
        for (const auto& top_account : trading_data->top_long_short_account_ratio()) {
          if (!pratio) {
            pratio = &(top_account_[std::make_pair(norm_symbol, trading_data->interval())]);
            pratio->SetInterval(trading_data->interval());
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetTopLongShortAccountRatio(pratio);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            if (is_snapshot) {
              pratio->SnapshotClear();
            } else {
              pratio->UpdateClear();
            }
          }

          auto* item = pratio->Add();
          item->timestamp = top_account.timestamp();
          item->long_short_ratio = top_account.long_short_ratio();
          item->long_account = top_account.long_();
          item->short_account = top_account.short_();
        }

        onTopLongShortAccountRatioFeed(update_);
      } else if (trading_data->top_long_short_position_ratio_size() > 0) {
        ::TradingData<TopLongShortPositionRatio>* pratio = nullptr;
        for (const auto& top_position : trading_data->top_long_short_position_ratio()) {
          if (!pratio) {
            pratio = &(top_position_[std::make_pair(norm_symbol, trading_data->interval())]);
            pratio->SetInterval(trading_data->interval());
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetTopLongShortPositionRatio(pratio);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            if (is_snapshot) {
              pratio->SnapshotClear();
            } else {
              pratio->UpdateClear();
            }
          }

          auto* item = pratio->Add();
          item->timestamp = top_position.timestamp();
          item->long_short_ratio = top_position.long_short_ratio();
          item->long_position = top_position.long_();
          item->short_position = top_position.short_();
        }

        onTopLongShortPositionRatioFeed(update_);
      } else if (trading_data->global_long_short_account_ratio_size() > 0) {
        ::TradingData<GlobalLongShortAccountRatio>* pratio = nullptr;
        for (const auto& global_account : trading_data->global_long_short_account_ratio()) {
          if (!pratio) {
            pratio = &(global_account_[std::make_pair(norm_symbol, trading_data->interval())]);
            pratio->SetInterval(trading_data->interval());
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetGlobalLongShortAccountRatio(pratio);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            if (is_snapshot) {
              pratio->SnapshotClear();
            } else {
              pratio->UpdateClear();
            }
          }

          auto* item = pratio->Add();
          item->timestamp = global_account.timestamp();
          item->long_short_ratio = global_account.long_short_ratio();
          item->long_account = global_account.long_();
          item->short_account = global_account.short_();
        }

        onGlobalLongShortAccountRatioFeed(update_);
      } else if (trading_data->taker_buy_sell_ratio_size() > 0) {
        ::TradingData<::TakerBuySellRatio> * pratio = nullptr;
        for (const auto& taker_vol : trading_data->taker_buy_sell_ratio()) {
          if (!pratio) {
            pratio = &(taker_volume_[std::make_pair(norm_symbol, trading_data->interval())]);
            pratio->SetInterval(trading_data->interval());
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetTakerBuySellRatio(pratio);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            if (is_snapshot) {
              pratio->SnapshotClear();
            } else {
              pratio->UpdateClear();
            }
          }

          auto* item = pratio->Add();
          item->timestamp = taker_vol.timestamp();
          item->buy_sell_ratio = taker_vol.buy_sell_ratio();
          item->buy_volume = taker_vol.buy_volume();
          item->sell_volume = taker_vol.sell_volume();
        }

        onTakerBuySellRatioFeed(update_);
      } else if (trading_data->open_interest_hist_size() > 0) {
        ::TradingData<::OpenInterest> * phist = nullptr;
        for (const auto& open_interest : trading_data->open_interest_hist()) {
          if (!phist) {
            phist = &(open_interest_hist_[std::make_pair(norm_symbol, trading_data->interval())]);
            phist->SetInterval(trading_data->interval());
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetOpenInterestHist(phist);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            if (is_snapshot) {
              phist->SnapshotClear();
            } else {
              phist->UpdateClear();
            }
          }

          auto* item = phist->Add();
          item->timestamp = open_interest.timestamp();
          item->open_interest_qty = open_interest.qty_float();
        }

        onOpenInterestHistFeed(update_);
      }
    }
  }

  std::vector<IProduct*> GetProducts() const override {
    std::vector<IProduct*> out;
    std::unordered_set<std::string> duplicate_check;
    for (const auto& p : reader_->products()) {
      std::string key = p->relative_norm() + p->absolute_norm();
      if(duplicate_check.count(key) == 0) {
        out.push_back(p);
        duplicate_check.insert(key);
      }
    }
    return out;
  }

 private:
  const MarketExchangeApiRecipe mear_;
  const FeedSystemConfig& config_;

  // states
  int64_t start_time_;
  std::unique_ptr<FastFeedReader> reader_;
  bool has_feed_ = false;
  std::map<std::string, std::unique_ptr<BookBuilder>> books_;
  Trade trade_{};
  Feed feed_;
  std::unordered_map<int, std::string> index_to_norm_symbol_;
  LiquidationOrder liquidation_order_{};
  OpenInterest open_interest_{};
  Index index_{};
  MarkPrice mark_price_{};
  FundingRate funding_rate_{};
  IvGreeks iv_greeks_{};
  Nav nav_{};
  Wallet wallet_{};
  MarketAnalysis market_analysis_{};
  // <symbol, interval> => klines
  std::map<std::pair<std::string, ::fastfeed::proto::KlineInterval>, Klines> klines_;
  // <symbol, interval> => trading data
  std::map<std::pair<std::string, ::fastfeed::proto::TradingDataInterval>,
      TradingData<TopLongShortAccountRatio>> top_account_;
  std::map<std::pair<std::string, ::fastfeed::proto::TradingDataInterval>,
      TradingData<TopLongShortPositionRatio>> top_position_;
  std::map<std::pair<std::string, ::fastfeed::proto::TradingDataInterval>,
      TradingData<GlobalLongShortAccountRatio>> global_account_;
  std::map<std::pair<std::string, ::fastfeed::proto::TradingDataInterval>,
      TradingData<TakerBuySellRatio>> taker_volume_;
  std::map<std::pair<std::string, ::fastfeed::proto::TradingDataInterval>,
      TradingData<OpenInterest>> open_interest_hist_;
  std::unique_ptr<FastFeedUpdate> update_ptr_;

  // cache
  // norm symbol => product
  std::map<std::string, std::unique_ptr<IProduct>> products_;
};

}  // namespace impl
using impl::FastFeedSubsystem;
using impl::FastFeedUpdate;
}  // namespace coin2::feed::fastfeed
